package org.hylexus.hadoop.hadoop_2_test.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;

public class HBaseTest1 {

	@Test
	@SuppressWarnings("deprecation")
	public void createTableUser() throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "192.168.164.128:2181");
		HBaseAdmin admin = new HBaseAdmin(conf);
		HTableDescriptor desc = new HTableDescriptor(TableName.valueOf("user"));
		desc.addFamily(new HColumnDescriptor("info").setMaxVersions(3));
		desc.addFamily(new HColumnDescriptor("extra").setMaxVersions(2));
		admin.createTable(desc);
		admin.close();
	}

	public Configuration getConf() {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "192.168.164.128:2181");
		return conf;
	}

	public static byte[] t(String s) {
		return Bytes.toBytes(s);
	}

	@SuppressWarnings("deprecation")
	@Test
	public void testPut() {
		try {
			Configuration conf = getConf();
			HTable table = new HTable(conf, "user");
			Put p1 = new Put(Bytes.toBytes("rk001"))
					.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("tom"))
					.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(100));

			Put p2 = new Put(Bytes.toBytes("rk002")).addColumn(t("info"), t("name"), t("cat"))
					.addColumn(t("extra"), t("desc"), t("this is a desc field"))
					.addColumn(t("info"), t("birth"), Bytes.toBytes("2015-02-02"));

			table.put(Arrays.asList(p1, p2));
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@SuppressWarnings("deprecation")
	@Test
	public void testBatchPut() {
		try {
			Configuration conf = getConf();
			HTable table = new HTable(conf, "user");
			List<Put> list = new ArrayList<>(10001);
			for (int i = 0; i < 100 * 10000; i++) {
				list.add(new Put(t("rk00" + (i + 3))).addColumn(t("info"), t("name"), t("tom" + (i + 3)))
						.addColumn(t("info"), t("salary"), t("$ " + 1000 + i)));
				if (i % 10000 == 0) {
					table.put(list);
					list = new ArrayList<>(10001);
				}
			}
			table.put(list);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@SuppressWarnings("deprecation")
	@Test
	public void testGet() {
		try {
			HTable table = new HTable(getConf(), "user");
			Get get = new Get(Bytes.toBytes("rk001"));
			Result result = table.get(get);
			byte[] value = result.getValue(t("info"), t("name"));
			System.err.println(Bytes.toString(value));
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@SuppressWarnings("deprecation")
	@Test
	public void testScan() {
		try {
			HTable table = new HTable(getConf(), "user");
			Scan scan = new Scan(t("rk001"), t("rk001000005"));
			ResultScanner resultScanner = table.getScanner(scan);
			if (resultScanner != null) {
				for (Result r : resultScanner) {
					System.err.println(r);
				}
				resultScanner.close();
			}
			table.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@SuppressWarnings("deprecation")
	@Test
	public void testDelete() {
		try {
			HTable table = new HTable(getConf(), "user");

			Delete delete = new Delete(t("rk001")).addColumn(t("info"), t("name"));
			table.delete(delete);
			table.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
